package com.rn.gypsophila.grpc;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.examples.calculate.CalculateProto;
import io.grpc.examples.calculate.CalculateServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
public class GrpcClient {

    private static final String DEFAULT_HOST = "localhost";
    private static final int DEFAULT_PORT = 50051;
    private ManagedChannel managedChannel;
    private CalculateServiceGrpc.CalculateServiceStub serviceStub;
    private Subject subject;

    public GrpcClient(String host, int port) {
        this.managedChannel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
        this.serviceStub = CalculateServiceGrpc.newStub(this.managedChannel);
        subject = new Subject();
    }

    public void shutdown() {
        if (managedChannel != null && !managedChannel.isShutdown()) {
            log.info("即将关闭Channel...");
            managedChannel.shutdown();
        }
    }

    public void getResult() {
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        StreamObserver<CalculateProto.Result> responseObserver = new StreamObserver<CalculateProto.Result>() {
            private int cnt = 0;

            @Override
            public void onNext(CalculateProto.Result result) {
                //此处直接打印结果，其他也可用回调进行复杂处理
                log.info("收到第{}次返回结果:{}", ++cnt, result.toString());
                subject.execute(result.getTxHashList());
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("调用出错:{}", throwable.getMessage());
                log.info("即将连接到其他节点...");
//                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("调用完成");
                countDownLatch.countDown();
            }
        };
        StreamObserver<CalculateProto.Value> requestObserver = serviceStub.getResult(responseObserver);
        requestObserver.onNext(CalculateProto.Value.newBuilder().setValue(10).build());

        //判断调用结束状态。如果整个调用已经结束，继续发送数据不会报错，但是会被舍弃
        if (countDownLatch.getCount() == 0) {
            return;
        }
        //异步请求，无法确保onNext与onComplete的完成先后顺序
        requestObserver.onCompleted();
        try {
            Runtime.getRuntime().addShutdownHook(new Thread(GrpcClient.this::shutdown));
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        System.out.println("*************************Getting result from server***************************");
        GrpcClient client = new GrpcClient(DEFAULT_HOST, DEFAULT_PORT);
        client.subject.addCallback("tt", new Callback() {
            @Override
            public void onSuccess() {
                log.info("执行回调->onSuccess");
            }

            @Override
            public void onFailure() {
                log.info("执行回调->onFailure");
            }
        });
        client.subject.addCallback("qq", new Callback() {
            @Override
            public void onSuccess() {
                log.info("执行回调->onSuccess");
            }

            @Override
            public void onFailure() {
                log.info("执行回调->onFailure");
            }
        });
        client.getResult();
    }

}